package io.kiki.sba.registry.server.clientInterface.push;


import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class ChangeProcessor {
    private static final Logger logger = LoggerFactory.getLogger(ChangeProcessor.class);

    @Autowired
    Config config;

    Map<String, Worker[]> dataCenterToWorkersMap = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        Worker[] workers = initWorkers();
        dataCenterToWorkersMap.putIfAbsent(config.getDataCenter(), workers);
    }

    private Worker[] initWorkers() {
        Worker[] workers = new Worker[config.getDataChangeFetchTaskWorkerSize()];
        for (int i = 0; i < workers.length; i++) {
            workers[i] = new Worker(config.getDataChangeDebouncingMillis(), config.getDataChangeMaxDebouncingMillis());
            ConcurrentUtils.createDaemonThread("ChangeExecutor-" + i, workers[i]).start();
        }
        return workers;
    }

    public void setWorkDelayTime(PushEfficiencyImproveConfig pushEfficiencyImproveConfig) {
        for (Map.Entry<String, Worker[]> entry : dataCenterToWorkersMap.entrySet()) {
            Worker[] workers = entry.getValue();
            if (workers == null) {
                return;
            }
            for (Worker work : workers) {
                work.setChangeTaskWorkDelay(pushEfficiencyImproveConfig);
            }
        }
    }

    boolean fireChange(String dataInfoId, ChangeHandler changeHandler, TriggerPushContext triggerPushContext) {
        ChangeKey changeKey = new ChangeKey(triggerPushContext.dataCenters(), dataInfoId);
        Worker worker = of(changeKey);
        return worker.commitChange(changeKey, changeHandler, triggerPushContext);
    }

    Worker of(ChangeKey changeKey) {
        String dataCenter = changeKey.dataCenters.stream().findFirst().get();
        Worker[] workers = dataCenterToWorkersMap.computeIfAbsent(dataCenter, k -> initWorkers());
        int n = (changeKey.hashCode() & 0x7fffffff) % workers.length;
        return workers[n];
    }

    interface ChangeHandler {
        boolean onChange(String dataInfoId, TriggerPushContext triggerPushContext);
    }

    static final class ChangeTask {
        final TriggerPushContext triggerPushContext;
        final ChangeKey changeKey;
        final ChangeHandler changeHandler;
        final long expireTimestamp;
        long expireDeadlineTimestamp;

        ChangeTask(ChangeKey changeKey, TriggerPushContext triggerPushContext, ChangeHandler changeHandler, long expireTimestamp) {
            this.changeKey = changeKey;
            this.changeHandler = changeHandler;
            this.triggerPushContext = triggerPushContext;
            this.expireTimestamp = expireTimestamp;
        }

        void doChange() {
            changeHandler.onChange(changeKey.dataInfoId, triggerPushContext);
        }


    }

    static final class Worker extends WakeUpLoopExecuteTask {
        // task sorted by expire probably
        final LinkedHashMap<ChangeKey, ChangeTask> map = new LinkedHashMap<>();
        int changeDebouncingMillis;
        int changeDebouncingMaxMillis;
        int changeTaskWaitingMillis = 100;

        Worker(int changeDebouncingMillis, int changeDebouncingMaxMillis) {
            this.changeDebouncingMillis = changeDebouncingMillis;
            this.changeDebouncingMaxMillis = changeDebouncingMaxMillis;
        }

        public void setChangeTaskWorkDelay(PushEfficiencyImproveConfig pushEfficiencyImproveConfig) {
            this.changeDebouncingMillis = pushEfficiencyImproveConfig.getChangeDebouncingMillis();
            this.changeDebouncingMaxMillis = pushEfficiencyImproveConfig.getChangeDebouncingMaxMillis();
            this.changeTaskWaitingMillis = pushEfficiencyImproveConfig.getChangeTaskWaitingMillis();
        }

        ChangeTask get(ChangeKey key) {
            synchronized (map) {
                return map.get(key);
            }
        }

        boolean commitChange(ChangeKey changeKey, ChangeHandler changeHandler, TriggerPushContext triggerPushContext) {
            final long now = System.currentTimeMillis();
            final ChangeTask changeTask = new ChangeTask(changeKey, triggerPushContext, changeHandler, now + changeDebouncingMillis);

            synchronized (map) {
                final ChangeTask exist = map.get(changeKey);
                if (exist == null) {
                    changeTask.expireDeadlineTimestamp = now + changeDebouncingMaxMillis;
                    map.put(changeKey, changeTask);
                    return true;
                }

                if (changeTask.triggerPushContext.smallerThan(exist.triggerPushContext)) {
                    return false;
                }
                changeTask.triggerPushContext.mergeVersion(exist.triggerPushContext);
                // compare with exist
                if (changeTask.expireTimestamp <= exist.expireDeadlineTimestamp) {
                    // not reach deadline, requeue to wait
                    changeTask.expireDeadlineTimestamp = exist.expireDeadlineTimestamp;
                    // merge change, merge tracetimes
                    changeTask.triggerPushContext.addTraceTime(exist.triggerPushContext.getFirstTraceTimes());
                    // tasks is linkedMap, must remove the exist first, then enqueue in the tail
                    map.remove(changeKey);
                    map.put(changeKey, changeTask);
                } else {
                    // reach deadline, could not requeue, use exist.expire as newTask.expire
                    exist.triggerPushContext.setExpectDatumVersion(changeTask.triggerPushContext.getExpectDatumVersion());
                }

                return true;
            }
        }

        ChangeTask getExpire() {
            final long now = System.currentTimeMillis();
            synchronized (map) {
                if (map.isEmpty()) {
                    return null;
                }
                Iterator<ChangeTask> iterator = map.values().iterator();
                final ChangeTask first = iterator.next();
                if (first.expireTimestamp <= now) {
                    iterator.remove();
                    return first;
                }
                return null;
            }
        }

        @Override
        public void _execute_() {
            for (; ; ) {
                final ChangeTask changeTask = getExpire();
                if (changeTask == null) {
                    break;
                }
                try {
                    changeTask.doChange();
                } catch (Throwable e) {
                    logger.error("failed to doChange, {}", changeTask);
                }
            }
        }

        @Override
        public int getWaitingMillis() {
            return changeTaskWaitingMillis;
        }
    }

    static final class ChangeKey {
        final String dataInfoId;
        final Set<String> dataCenters;

        ChangeKey(Set<String> dataCenters, String dataInfoId) {
            this.dataCenters = dataCenters;
            this.dataInfoId = dataInfoId;
        }


        @Override
        public boolean equals(Object o) {
            if (this == o)
                return true;
            if (o == null || getClass() != o.getClass())
                return false;
            ChangeKey changeKey = (ChangeKey) o;
            return Objects.equals(dataInfoId, changeKey.dataInfoId) && Objects.equals(dataCenters, changeKey.dataCenters);
        }

        @Override
        public int hashCode() {
            return Objects.hash(dataInfoId, dataCenters);
        }
    }
}
